<?php

interface RedisStorageInterface
{
    // connect
    public function connect($ip, $port, $timeout);

    // basic methods
    public function exists($k);
    public function del($k);
    public function rename($old, $new);
    public function expire($k, $ttl);
    public function expireAt($k, $ts);
    public function type($k);
    public function ttl($k);
    public function keys($pattern = '*');
    public function getKeys($pattern = '*');

    // string operations
    public function set($k, $v);
    public function get($k);
    public function mGet($ks);
    public function mSet($vs);
    public function incr($k);
    public function decr($k);

    // atomic operations
    public function getSet($k, $new);
    public function setEx($k, $v, $time);
    public function setNx($k, $v);
    public function mSetNx($vs);

    // set operations
    public function sAdd($k, $m);
    public function sRem($k, $m);
    public function sPop($k);
    public function sRandMember($k);
    public function sCard($k);
    public function sMembers($k);
    public function isMember($k, $m);
    public function sMove($src, $dst, $m);
    public function sUnion();
    public function sDiff();
    public function sInter();
    public function sUnionStore();
    public function sDiffStore();
    public function sInterStore();
    public function msAdd($ks, $ms);

    // list operations
    public function lPush($k, $v);
    public function rPush($k, $v);
    public function lPop($k);
    public function rPop($k);
    public function llen($k);
    public function lRange($k, $s, $e);
    public function lTrim($k, $s, $e);
    public function lSet($k, $i, $v);
    public function lGet($k, $i);
    public function lRem($k, $c, $v);
    public function sort($k,$a=array());

    // hash operations
    public function hDel($k, $f);
    public function hGetAll($k);
    public function hLen($k);
    public function hSet($k, $f, $v);
    public function hExists($k, $f);
    public function hIncrBy($k, $f, $i = 1);
    public function hDecrBy($k, $f, $i = 1);
    public function hMGet($k, $m);
    public function hSetNx($k, $f, $v);
    public function hGet($k, $f);
    public function hKeys($k);
    public function hMSet($k, $m);
    public function hVals($k);

    // sorted sets
    public function zAdd($k, $s, $m);
    public function zCard($k);
    public function zCount($k, $min, $max);
    public function zIncrBy($k, $i, $m);
    public function zInterStore($dk, $keys, $weights, $aggr);
    public function zRange($k, $start, $stop, $with_score = false);
    public function zRangeByScore($k, $min, $max, $options = array());
    public function zRank($k, $m);
    public function zRem($k, $m);
    public function zRemRangeByRank($k, $start, $stop);
    public function zRemRangeByScore($k, $min, $max);
    public function zRevRange($k, $start, $stop, $with_score = false);
    public function zRevRangeByScore($k, $min, $max, $options = array());
    public function zRevRank($k, $m);
    public function zScore($k, $m);
    public function zUnionStore($dk, $keys, $weights, $aggr);

    // database operations
    public function select($i);
    public function move($i);
    public function dbSize();
    public function save();
    public function bgSave();
    public function lastSave();
    public function flushDB();
    public function flushAll();
    public function info();
    public function multi($type);
    public function exec();
    public function discard();
    public function ping();
}

class RedisFactory
{
    static private $instances;
    static private $no_configs = array();

    static public function getInstances()
    {
        return self::$instances;
    }

    /**
     * @static
     * @param string $redis_name
     * @param string $group_name
     * @param null $new_instance
     * @return PHPRedisStorage
     */
    static public function getInstance($redis_name = DEFAULT_REDIS_NAME, $group_name = '', $new_instance = null)
    {
        if (isset(self::$instances[$redis_name]) && is_object(self::$instances[$redis_name]))
        {
            self::$instances[$redis_name]->setRedisName($redis_name);
            self::$instances[$redis_name]->setGroupName($group_name);
            self::$instances[$redis_name]->setNewInstance($new_instance);
        }
        else
        {
            self::$instances[$redis_name] = new self($redis_name, $group_name, $new_instance);
        }

        return self::$instances[$redis_name];
    }

    static private $log = array();
    static private $slave_node = array();
    private $config;
    private $node_config;
    private $redis_name;
    private $group_name;
    private $new_instance;
    private $write_methods = array(
        'del', 'rename', 'expire', 'expireAt', 'set', 'mSet', 'incr', 'decr', 'getSet', 'setEx', 'setNx', 'mSetNx',
        'sAdd', 'msAdd', 'sRem', 'sPop', 'sMove', 'lPush', 'rPush', 'lPop', 'rPop', 'lTrim', 'lSet', 'lRem', 'multi', 'multiExec', 'exec', 'flushAll', 'flushDB', 'hMSetEx', 'hMSet', 'hSetNx', 'hDecrBy', 'hIncrBy', 'hSet', 'hDel', 'sort', 'zAdd', 'zIncrBy', 'zInterStore', 'zRem', 'zUnionStore', 'zRemRangeByRank', 'zRemRangeByScore', 'sUnionStore', 'sDiffStore', 'sInterStore',
    );

    function __construct($redis_name, $group_name = '', $new_instance = null)
    {
        $this->setRedisName($redis_name);
        $this->setGroupName($group_name);
        $this->setNewInstance($new_instance);
    }

    public function setRedisName($redis_name)
    {
        $this->redis_name = $redis_name;
    }

    public function setGroupName($group_name)
    {
        $this->group_name = $group_name;
    }

    public function setNewInstance($new_instance)
    {
        $this->new_instance = $new_instance;
    }

    private function doLog($func, $time, $ret, $params = array())
    {
        if (DEBUG == false)
            return;
        $traces = array();
        foreach (array_slice(debug_backtrace(), 2) as $trace)
        {
            $traces[] = array(
                'file'  =>  $trace['file'],
                'line'  =>  $trace['line'],
                'func'  =>  $trace['function'],
            );
        }

        self::$log[] = array(
            'ret'   =>  $ret === false ? false : true,
            'redis' =>  $this->redis_name,
            'group' =>  $this->group_name,
            'func'  =>  $func,
            'params'=>  $params,
            'trace' =>  $traces,
            'config'=>  $this->node_config,
            'time'  =>  $time,
        );
    }

    public function getLog()
    {
        return self::$log;
    }

    private function initConfig($func, $params)
    {
        global $config_arr;
        $config = $config_arr;

        if (isset($config['redis'][$this->redis_name]))
        {
            // had config info
            $this->config = $config['redis'][$this->redis_name];
        }
        else
        {
            // no config info, use default here
            $this->config = $config['redis'][DEFAULT_REDIS_NAME];
        }

        // 如果没有指定某个节点，用HASH指定
        if ($this->group_name && isset($this->config[$this->group_name]))
        {
            $group = $this->config[$this->group_name];
        }
        else
        {
            $groups = array_keys($this->config);

            if (count($groups) > 1)
            {
                // 通过CONSISTENT HASH找到对应的REDIS服务器
                $hash = new ConsistentHash();
                // 1.载入该应用所有节点组
                $hash->addTargets($groups);

                // 2.根据KEY寻找对应节点
                if ($params[0] && is_string($params[0]))
                {
                    $key = $params[0];
                    $this->group_name = $hash->lookup($key);
                    $group = $this->config[$this->group_name];
                }
            }

            if (! $this->group_name)
            {
                // 否则取第一个节点
                $this->group_name = $groups[0];
                $group = $this->config[$this->group_name];
            }
        }

        if (in_array($func, $this->write_methods))
        {
            // 写操作
            $config = $group['master'];
        }
        else
        {
            // 读操作
            $config = null;
            if (isset(self::$slave_node[$this->redis_name]) && self::$slave_node[$this->redis_name])
            {
                $config = self::$slave_node[$this->redis_name];
            }
            else
            {
                if (isset($group['slave']))
                {
                    $nodes = array();
                    foreach ((array)$group['slave'] as $node)
                    {
                        if (! in_array($node, self::$no_configs))
                        {
                            $nodes[] = $node;
                        }
                    }
                    $config = $nodes[array_rand($nodes)];
                    self::$slave_node[$this->redis_name] = $config;
                }
            }
            if (! $config)
            {
                $config = $group['master'];
            }

//            if ($func != 'sort' || empty($nodes))
//            {
//            	$nodes[] = $group['master'];
//            }
//            $config = $nodes[array_rand($nodes)];
        }

        return $config;
    }

    protected function getRedisObj($func, $params, $n = 1)
    {
        if ($n > 3)
        {
            l('找不到可用REDIS节点', 'redis_connect');
            return false;
        }

        $this->node_config = $this->initConfig($func, $params);
        try
        {
            return PHPRedisStorage::getInstance($this->node_config, $this->new_instance);
        }
        catch(RedisException $e)
        {
            // 当前配置出现问题，剔除后重新选择
            l($this->node_config . ' 故障' . $n, 'redis_connect');
            self::$no_configs[] = $this->node_config;
            return $this->getRedisObj($func, $params, ++$n);
        }
    }

    protected function callFunc($func, $params, $n = 1)
    {
        if ($n > 1)
        {
//            l('命令执行失败', 'redis_runtime');
            return false;
        }

        $start_time = microtime(true);
        try
        {
            return call_user_func_array($func, $params);
        }
        catch(RedisException $e)
        {
            l(array('error'=>$e, 'time'=>(microtime(true) - $start_time)), 'redis_runtime');
            return $this->callFunc($func, $params, ++$n);
        }
    }

    public function __call($func, $params)
    {
        $redis = $this->getRedisObj($func, $params);

        $start_time = microtime(true);
        $ret = $this->callFunc(array($redis, $func), $params);
        $end_time = microtime(true);
        $exec_time = $end_time - $start_time;

        if (! defined('LOG_REDIS_LONGTIME'))
            define('LOG_REDIS_LONGTIME', false);

        if (LOG_REDIS_LONGTIME && $exec_time > LOG_REDIS_LONGTIME)
            l(array('func'=>$func, 'params'=>$params, 'time'=>$exec_time), 'redis_longtime');
        if (defined('RUNNING_MODE') && RUNNING_MODE == 'dev')
            $this->doLog($func, $exec_time, $ret, $params);

        return $ret;
    }
}

if (!class_exists('Redis'))
{
    class Redis{}
}

class PHPRedisStorage extends Redis implements RedisStorageInterface
{
    private $multi_exec_times = 1000;

    private $is_piping = false;
    private $connect_time = 0;
    private static $instances = array();

    public static function getInstance($config, $new_instance = null)
    {
        list($ip, $port) = explode(':', $config);
        if (!isset(self::$instances[$config]))
        {
            self::$instances[$config] = new self();

            $ret = self::$instances[$config]->_connect($ip, $port);
            if (! $ret)
            {
                throw new RedisException('Connected Faild');
            }
        }
        else if ((is_null($new_instance) && self::$instances[$config]->is_piping) || $new_instance)
        {
            // 返回新的实例
            $obj = new self();
            $ret = self::$instances[$config]->_connect($ip, $port);
            if (! $ret)
            {
                throw new RedisException('Connected Faild');
            }
            return $obj;
        }
        return self::$instances[$config];
    }

    public function getConnTime()
    {
        return $this->connect_time;
    }
    /*
        function __construct($ip, $port, $timeout = 0.5)
        {
            $this->_connect($ip, $port, $timeout);
        }
    */
    protected function _connect($ip, $port = '6379', $timeout = 5, $n = 1)
    {
        if ($n > 3)
        {
            return false;
        }
        try
        {
            $start_time = microtime(true);
            $this->connect($ip, $port, $timeout);
            $end_time = microtime(true);
            $this->connect_time = round($end_time - $start_time, 3);
            return true;
        }
        catch(RedisException $e)
        {
            l($ip . ':' . $port . ' 连接失败' . $n, 'redis_connect');
            return $this->_connect($ip, $port, $timeout, ++$n);
        }
    }

    public function connect($ip, $port, $timeout)
    {
        parent::connect($ip, $port, $timeout);
    }

    // basic methods
    public function exists($k)
    {
        return parent::exists($k);
    }
    public function del($k)
    {
        return parent::delete($k);
    }

    public function rename($old, $new)
    {
        return parent::renameKey($old, $new);
    }
    public function expire($k, $ttl)
    {
        return parent::expire($k, $ttl);
    }
    public function expireAt($k, $ts)
    {
        return parent::expireAt($k, $ts);
    }
    public function type($k)
    {
        return parent::type($k);
    }
    public function ttl($k)
    {
        return parent::ttl($k);
    }
    public function keys($pattern = '*')
    {
        return parent::getKeys($pattern);
    }
    public function getKeys($pattern = '*')
    {
        return parent::getKeys($pattern);
    }

    // string operations
    public function set($k, $v)
    {
        return parent::set($k, $v);
    }
    public function get($k)
    {
        return parent::get($k);
    }
    public function mGet($ks)
    {
        return parent::getMultiple($ks);
    }
    public function mSet($vs)
    {
        $ret = parent::mset($vs);
        return $ret;
    }
    public function incr($k)
    {
        return parent::incr($k);
    }
    public function decr($k)
    {
        return parent::decr($k);
    }

    // atomic operations
    public function getSet($k, $new)
    {
        return parent::getSet($k, $new);
    }

    public function setEx($k, $v, $time)
    {
        parent::setex($k, $time, $v);
    }

    public function setNx($k, $v)
    {
        return parent::setnx($k, $v);
    }

    public function mSetNx($vs)
    {
        die('not implemented yet');
    }

    // set operations
    public function sAdd($k, $m)
    {
        $params = (array)$m;
        array_unshift($params, $k);
        return call_user_func_array(array('parent', 'sAdd'), $params);
//        return parent::sAdd($k, $m);
    }
    public function sRem($k, $m)
    {
        $params = (array)$m;
        array_unshift($params, $k);
        return call_user_func_array(array('parent', 'sRemove'), $params);
//        return parent::sRemove($k, $m);
    }
    public function sPop($k)
    {
        return parent::sPop($k);
    }
    public function sRandMember($k)
    {
        return parent::sRandMember($k);
    }
    public function sCard($k)
    {
        return parent::sSize($k);
    }
    public function sMembers($k)
    {
        return parent::sMembers($k);
    }
    public function isMember($k, $m)
    {
        return parent::sContains($k, $m);
    }
    public function sMove($src, $dst, $m)
    {
        return parent::sMove($src, $dst, $m);
    }
    public function sUnion()
    {
        $args = func_get_args();
        return call_user_func_array(array('parent', 'sUnion'), $args);
    }
    public function sDiff()
    {
        $args = func_get_args();
        return call_user_func_array(array('parent', 'sDiff'), $args);
    }
    public function sInter()
    {
        $args = func_get_args();
        return call_user_func_array(array('parent', 'sInter'), $args);
    }

    public function sUnionStore()
    {
        $args = func_get_args();
        return call_user_func_array(array('parent', 'sUnionStore'), $args);
    }
    public function sDiffStore()
    {
        $args = func_get_args();
        return call_user_func_array(array('parent', 'sDiffStore'), $args);
    }
    public function sInterStore()
    {
        $args = func_get_args();
        return call_user_func_array(array('parent', 'sInterStore'), $args);
    }
    private function _sAdd($km, $redis)
    {
        $redis->sAdd($km['k'], $km['m']);
    }
    public function msAdd($ks, $ms)
    {
        $km = array();
        if (count($ks) < count($ms))
        {
            $a1 = $ks;
            $a2 = $ms;
            $k = 'a';
            $m = 'aa';
        }
        else
        {
            $a1 = $ms;
            $a2 = $ks;
            $k = 'aa';
            $m = 'a';
        }
        foreach ((array)$a1 as $a)
        {
            foreach ((array)$a2 as $aa)
            {
                $km[] = array('k' => $$k, 'm' => $$m);
            }
        }
        return $this->multiExec(array(& $this, '_sAdd'), $km);
    }


    // sorted sets
    public function zAdd($k, $s, $m)
    {
        $params = array($k);
        $s = (array)$s;
        $m = (array)$m;
        foreach ($s as $n => $i)
        {
            $params[] = $i;
            $params[] = $m[$n];
        }
        return call_user_func_array(array('parent', 'zAdd'), $params);
//      return parent::zAdd($k, $s, $m);
    }
    public function zCard($k)
    {
        return parent::zCard($k);
    }
    public function zCount($k, $min, $max)
    {
        return parent::zCount($k, $min, $max);
    }
    public function zIncrBy($k, $i, $m)
    {
        return parent::zIncrBy($k, $i, $m);
    }
    public function zInterStore($dk, $keys, $weights, $aggr)
    {
        return parent::zInter($dk, $keys, $weights, $aggr);
    }
    public function zRange($k, $start, $stop, $with_score = false)
    {
        return parent::zRange($k, $start, $stop, $with_score);
    }
    public function zRangeByScore($k, $min, $max, $options = array())
    {
        return parent::zRangeByScore($k, $min, $max, $options);
    }
    public function zRank($k, $m)
    {
        return parent::zRank($k, $m);
    }
    public function zRem($k, $m)
    {
        $params = (array)$m;
        array_unshift($params, $k);
        return call_user_func_array(array('parent', 'zDelete'), $params);
//        return parent::zDelete($k, $m);
    }
    public function zRemRangeByRank($k, $start, $stop)
    {
        return parent::zRemRangeByRank($k, $start, $stop);
    }
    public function zRemRangeByScore($k, $min, $max)
    {
        return parent::zRemRangeByScore($k, $min, $max);
    }
    public function zRevRange($k, $start, $stop, $with_score = false)
    {
        return parent::zReverseRange($k, $start, $stop, $with_score);
    }
    public function zRevRangeByScore($k, $min, $max, $options = array())
    {
        return parent::zRevRangeByScore($k, $min, $max, $options);
    }
    public function zRevRank($k, $m)
    {
        return parent::zRevRank($k, $m);
    }
    public function zScore($k, $m)
    {
        return parent::zScore($k, $m);
    }
    public function zUnionStore($dk, $keys, $weights, $aggr)
    {
        return parent::zUnion($dk, $keys, $weights, $aggr);
    }


    // list operations
    public function lPush($k, $v)
    {
        $params = (array)$v;
        array_unshift($params, $k);
        return call_user_func_array(array('parent', 'lPush'), $params);
//        return parent::lPush($k, $v);
    }
    public function rPush($k, $v)
    {
        $params = (array)$v;
        array_unshift($params, $k);
        return call_user_func_array(array('parent', 'rPush'), $params);
//        return parent::rPush($k, $v);
    }
    public function lPop($k)
    {
        return parent::lPop($k);
    }
    public function rPop($k)
    {
        return parent::rPop($k);
    }
    public function llen($k)
    {
        return parent::lSize($k);
    }
    public function lRange($k, $s, $e)
    {
        return parent::lRange($k, $s, $e);
    }
    public function lTrim($k, $s, $e)
    {
        return parent::listTrim($k, $s, $e);
    }
    public function lSet($k, $i, $v)
    {
        return parent::lSet($k, $i, $v);
    }
    public function lGet($k, $i)
    {
        return parent::lGet($k, $i);
    }
    public function lRem($k, $c, $v)
    {
        return parent::lRemove($k, $v, $c);
    }
    public function sort($k,$a=array())
    {
        return parent::sort($k, $a);
    }

    // hash operations
    public function hDel($k, $f)
    {
        $params = (array)$f;
        array_unshift($params, $k);
        return call_user_func_array(array('parent', 'hDel'), $params);
//        return parent::hDel($k, $f);
    }
    public function hGetAll($k)
    {
        return parent::hGetAll($k);
    }
    public function hLen($k)
    {
        return parent::hLen($k);
    }
    public function hSet($k, $f, $v)
    {
        return parent::hSet($k, $f, $v);
    }
    public function hExists($k, $f)
    {
        return parent::hExists($k, $f);
    }
    public function hIncrBy($k, $f, $i = 1)
    {
        return parent::hIncrBy($k, $f, $i);
    }
    public function hDecrBy($k, $f, $i = 1)
    {
        return $this->hIncrBy($k, $f, -$i);
    }
    /**
     * 当$m中含有HASH结构中不存在的字段时，该字段会返回一个空值
     * 输入、输出参数（数组）个数保持不变
     */
    public function hMGet($k, $m)
    {
        return parent::hMGet($k, $m);
    }
    public function hSetNx($k, $f, $v)
    {
        if ($this->hExists($k, $f))
        {
            return false;
        }
        else
        {
            $this->hSet($k, $f, $v);
            return true;
        }
    }
    public function hGet($k, $f)
    {
        return parent::hGet($k, $f);
    }
    public function hKeys($k)
    {
        return parent::hKeys($k);
    }
    public function hMSet($k, $m)
    {
        return parent::hMSet($k, $m);
    }
    public function hVals($k)
    {
        return parent::hVals($k);
    }

    public function hMSetEx($k, $m, $expire)
    {
        $this->multi(Redis::PIPELINE);
        $this->hMSet($k, $m);
        $this->expire($k, $expire);
        return $this->exec();
    }

    // database operations
    public function select($i)
    {
        return parent::select($i);
    }
    public function move($i)
    {
        return parent::move($i);
    }
    public function dbSize()
    {
        return parent::dbSize();
    }
    public function save()
    {
        return parent::save();
    }
    public function bgSave()
    {
        return parent::bgsave();
    }
    public function lastSave()
    {
        return parent::lastSave();
    }
    public function flushDB()
    {
        return parent::flushDB();
    }
    public function flushAll()
    {
        return parent::flushAll();
    }
    public function info()
    {
        return parent::info();
    }
    public function multi($type)
    {
        $this->is_piping = true;
        return parent::multi($type);
    }
    public function exec()
    {
        $ret = parent::exec();
        $this->is_piping = false;
        return $ret;
    }
    public function discard()
    {
        $ret = parent::discard();
        $this->is_piping = false;
        return $ret;
    }

    public function multiExec2()
    {
        $args = func_get_args();
        return call_user_func_array(array(& $this, 'multiexec'), $args);
    }

    public function multiExec()
    {
        $args = func_get_args();

        $func = array_shift($args);
        $var  = array_shift($args);
        $num  = $this->multi_exec_times;


        $this->is_piping = true;
        $ret = array();

        if (is_callable($func))
        {
            $redis = $this->multi(Redis::PIPELINE);
            $var = (array)$var;
            $var_ct = count($var);
            $last_exec = true;
            foreach ($var as $n => $item)
            {
                $params = array_merge(array($item), $args, array($redis, $n));

                call_user_func_array($func, $params);
                if (($n + 1) % $num == 0)
                {
                    $last_exec = false;
                    $ret = array_merge($ret, $this->exec());
                    if ($n != ($var_ct - 1))
                    {
                        $redis = $this->multi(Redis::PIPELINE);
                        $last_exec = true;
                    }
                }
            }
            if ($last_exec)
            {
                $ret = array_merge($ret, $this->exec());
            }

            return $ret;
        }
        else
        {
            return false;
        }
    }

    public function ping()
    {
        return parent::ping();
    }

    public function setMultiExecTimes($times)
    {
        $this->multi_exec_times = $times;
    }

    public function getMultiExecTimes()
    {
        return $this->multi_exec_times;
    }
}

interface MessageQueueInterface
{
    public function push($v);
    public function pop();
    public function getName();
    public function setName($name);
}

class RedisMQ implements MessageQueueInterface
{
    private $name;
    private $storage;

    static function newMQ($name, $redis = DEFAULT_REDIS_NAME)
    {
        return new self($name, $redis);
    }

    function __construct($name, $redis)
    {
        $this->storage = RedisFactory::getInstance($redis);
        $this->name = $name;
    }

    public function push($v, $d = 'r')
    {
        if ($d == 'r')
            $this->storage->rPush($this->name, $v);
        else
            $this->storage->lPush($this->name, $v);
    }

    public function pop($d = 'l')
    {
        if ($d == 'l')
            return $this->storage->lPop($this->name);
        else
            return $this->storage->rPop($this->name);
    }

    public function getName()
    {
        return $this->name;
    }

    public function setName($name)
    {
        $this->name = $name;
    }

    public function ping()
    {
        return $this->storage->ping();
    }
}
?>